package com.google.code.coma.logistics;

import com.google.code.coma.Coma;
import com.google.code.coma.domain.Computer;
import com.google.code.coma.domain.LifeCycleStep;
import com.google.code.coma.domain.Order;
import com.google.code.coma.space.ComputerSelectorCallback;
import com.google.code.coma.space.OrderSelectorCallback;
import com.google.code.coma.space.SpaceSelector;
import com.google.code.coma.space.util.SpaceManager;
import com.google.code.coma.worker.Worker;
import com.sun.org.apache.xpath.internal.operations.Or;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mozartspaces.core.*;

import java.net.URI;
import java.util.List;

/**
 * @author Bernhard Nickel, e0925384@student.tuwien.ac.at
 */
public class SpaceLogisticsWorker extends AbstractLogisticsWorker implements Worker {
    private static final Log log = LogFactory.getLog(SpaceLogisticsWorker.class);


	private MzsCore core;

	private Capi capi;

	private ContainerReference computers;
    private ContainerReference orders;

	private Thread executionThread;

	private URI spaceUri;

	public SpaceLogisticsWorker(String factoryId) {
		super(factoryId);

		core = DefaultMzsCore.newInstance(0);
		capi = new Capi(core);
		spaceUri = URI.create("xvsm://localhost:4242");
		try {
			computers = SpaceManager.getOrCreateNamedContainer(spaceUri, factoryId + "_" + Coma.COMPUTER_CONTAINER, capi);
            orders = SpaceManager.getOrCreateNamedContainer(spaceUri, factoryId + "_" + Coma.ORDER_CONTAINER, capi);
		} catch (MzsCoreException e) {
			core.shutdown(false);
			throw new RuntimeException("Unable to get container", e);
		}
	}


	@Override
	public void startWork() {
		executionThread = new Thread(new Runnable() {
			@Override
			public void run() {
				while (!Thread.currentThread().isInterrupted()) {
					TransactionReference tx;
					try {
						tx = capi.createTransaction(MzsConstants.TransactionTimeout.INFINITE, spaceUri);
					} catch (MzsCoreException e) {
						log.error("Unable to create transaction", e);
						return;
					}
					try {
                        List<Computer> takenComputers = capi.take(computers, new SpaceSelector(new ComputerSelectorCallback(LifeCycleStep.VERIFIED)), MzsConstants.RequestTimeout.INFINITE, tx);
						Computer computer = takenComputers.get(0);
                        if (log.isInfoEnabled()) {
							log.info("Finishing computer " + computer);
						}

                        computer.changeState(LifeCycleStep.FINISHED, getWorkerId());

                        if (computer.getOrder() != null) {
                            List<Computer> orderComputers = capi.read(computers, new SpaceSelector(new ComputerSelectorCallback(computer.getOrder(), LifeCycleStep.FINISHED)), MzsConstants.RequestTimeout.INFINITE, tx);

                            if ((orderComputers.size() + 1) == computer.getOrder().getSizeOfOrder()) {
                                if (log.isInfoEnabled()) {
                                    log.info("Finishing order " + computer.getOrder());
                                }

                                List<Order> takenOrders = capi.take(orders, new SpaceSelector(new OrderSelectorCallback(computer.getOrder())), MzsConstants.RequestTimeout.INFINITE, tx);
                                Order order = takenOrders.get(0);
                                order.finish();
                                capi.write(new Entry(order), orders, MzsConstants.RequestTimeout.INFINITE, tx);
                            }
                        }

						capi.write(new Entry(computer), computers, MzsConstants.RequestTimeout.INFINITE, tx);
						capi.commitTransaction(tx);
					} catch (MzsCoreException e) {
						log.error("Space exception", e);
						try {
							capi.rollbackTransaction(tx);
						} catch (MzsCoreException e1) {
							log.error("Error in transaction rollback", e);
						}
						Thread.currentThread().interrupt();
					}
				}
			}
		});
		executionThread.start();
	}

	@Override
	public void stopWork() {
		if (executionThread != null) {
			executionThread.interrupt();
		}
	}

    @Override
    public boolean isActive() {
        return executionThread.isAlive();
    }
}
